package com.csx.governance.traffic.mq.rocketmq;

import cn.hutool.core.util.ObjectUtil;
import com.csx.governance.traffic.core.TrafficGovernanceRuleLocator;
import com.csx.governance.traffic.core.metadata.ServiceMetadataLocator;
import com.csx.governance.traffic.mq.rocketmq.hook.TrafficConsumeMessageHook;
import com.csx.governance.traffic.mq.rocketmq.model.RocketMqServiceRule;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MQPullConsumer;
import org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.ApplicationContext;

/**
 * 可以在此处修改消费者的订阅逻辑
 * @author cuisongxu
 * @date 2023/2/28 16:25
 */
public class TrafficMQConsumerBeanPostProcessor implements BeanPostProcessor {

    private static final Logger logger = LoggerFactory.getLogger(TrafficMQConsumerBeanPostProcessor.class);

    private final ApplicationContext applicationContext;
    private final TrafficGovernanceRuleLocator<RocketMqServiceRule> rocketMqGovernanceRuleLocator;
    private final ServiceMetadataLocator serviceMetadataLocator;
    private final String localApplicationName;

    public TrafficMQConsumerBeanPostProcessor(ApplicationContext applicationContext,
                                              TrafficGovernanceRuleLocator<RocketMqServiceRule>  rocketMqGovernanceRuleLocator,
                                              ServiceMetadataLocator serviceMetadataLocator) {
        this.applicationContext = applicationContext;
        this.rocketMqGovernanceRuleLocator = rocketMqGovernanceRuleLocator;
        this.serviceMetadataLocator = serviceMetadataLocator;
        localApplicationName = applicationContext.getEnvironment().getProperty("spring.application.name");
    }

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        // TODO 针对两种不同的消费者做特殊处理
        if(bean instanceof DefaultRocketMQListenerContainer) {
            DefaultRocketMQListenerContainer defaultRocketMQListenerContainer = (DefaultRocketMQListenerContainer) bean;
            DefaultMQPushConsumer mqPushConsumer = defaultRocketMQListenerContainer.getConsumer();
            TrafficConsumeMessageHook trafficConsumeMessageHook = applicationContext.getBean(TrafficConsumeMessageHook.class);
            mqPushConsumer.getDefaultMQPushConsumerImpl().registerConsumeMessageHook(trafficConsumeMessageHook);
            // 判断是否需要加入灰度
            if(!isNeedJoinGray()) {
                return BeanPostProcessor.super.postProcessAfterInitialization(bean, beanName);
            }

            // 获取灰度标签后缀
            String grayTagSuffix = getGrayTagSuffix();
            try {
                // 修改消费者组
                // 这里为什么要修改消费者组呢?
                // 主要是为了让同一应用不同版本的消费者处于不同的消费者组中, 因为如果同一应用不同版本的消费者如果在同一个消费者组中，
                // 那么会导致消息在同一个消费者组中进行负载，从而导致不是所有的消息路由到同一实例中进行过滤，这样会导致大量消息重发甚至重发多次。
                // 例如, 对于一个 green消息，green泳道中存在消费者，因此green消息只能被green消费者消费。如果 base, green, blue消费者全部在同一个消费者组中,
                // 那么green会随机路由到base, green, blue消费者中任意一个，base, blue是不能消费green消息的,如果不能正确路由到green消费者，那么green消息会不断重发导致一条消息消费会大大延迟。
                // 如果把 base, green, blue消费者设置在不同的消费者组，green消息就能同期进入到 base, green, blue 三个消费者组中，其中base, blue消费者丢弃green消息，green消费者消费该消息
                // 此时green消息只需要路由一次就能够被正确的green消费者消费，从而保证消息消费的延迟问题。
                String consumerGroup = defaultRocketMQListenerContainer.getConsumerGroup();
                consumerGroup = consumerGroup + "_" + grayTagSuffix;
                defaultRocketMQListenerContainer.setConsumerGroup(consumerGroup);
                mqPushConsumer.setConsumerGroup(consumerGroup);

            } catch (Exception e) {
                e.printStackTrace();
            }
        } else if(bean instanceof MQPullConsumer) {
            // TODO 该类实现类被遗弃，可以暂时不实现
            logger.error("use MQPullConsumer");
        }
        return BeanPostProcessor.super.postProcessAfterInitialization(bean, beanName);
    }

    private boolean isNeedJoinGray() {
        RocketMqServiceRule rule = rocketMqGovernanceRuleLocator.locateRule(localApplicationName);
        return ObjectUtil.isNotNull(rule) && rule.getEnabled();
    }

    private String getGrayTagSuffix() {
        RocketMqServiceRule rule = rocketMqGovernanceRuleLocator.locateRule(localApplicationName);
        String tagKey = rule.getTag();
        return serviceMetadataLocator.get(tagKey);
    }
}
